package com.shujia.sql

import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo03DSL {
  def main(args: Array[String]): Unit = {
    /**
     * DSL：特定领域语言，在SparkSQL的场景中主要用于基于DF完成数据分析
     * 包括两种表达方式：
     * 1、字符串表达式 df.where("age > 22")
     * 2、列表达式 df.where($"age" > 22)
     */
    /**
     * SQL中有哪些常见操作：
     * SELECT 函数(字段)
     * FROM tb
     * where 过滤
     * group by 分组 + 聚合函数
     * having 分组之后的过滤
     * order by 排序
     * limit 限制返回的条数
     *
     * 多表之间的连接操作：
     * join关联：
     * inner join 内连接
     * outer join 外连接：
     * left join 左外连接
     * right join 右外连接
     * full join 全外连接
     * union联接：
     * union 去重
     * union all 不会去重
     *
     * 函数：
     * 字符串函数
     * 数值函数
     * 条件函数
     * 窗口函数
     * 日期函数
     *
     */
    val spark: SparkSession = SparkSession
      .builder()
      .appName(this.getClass.getSimpleName)
      .master("local")
      .config("spark.sql.shuffle.partitions", "2")
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._

    val stuDF: DataFrame = spark.read.format("csv").option("sep", ",").schema("id String,name String,age Int,gender String,clazz String")
      .load("spark/data/stu/students.txt")



    // select 提取需要的字段 或者是 使用函数对字段进行处理 并且取别名
    stuDF.select("id", "name", "age")
    //      .show()
    stuDF.select($"id", $"name", $"age" as "Age")
    //      .show()

    // where 过滤
    // 过滤出age大于23的学生
    stuDF.where("age > 23")
    //      .show()
    stuDF.where($"age" > 23)
    //      .show()

    // groupBy 分组+聚合
    // 会有Shuffle出现，导致分区数过多，进而导致任务数过多
    // 可以通过参数：spark.sql.shuffle.partitions进行设置 默认200
    // 统计班级人数
    stuDF.groupBy($"clazz").agg(count($"id") as "cnt")
    //      .show()

    // 统计班级人数 并将人数大于100的班级过滤出来
    // 直接在聚合之后使用where进行过滤即可，不需要考虑执行的先后顺序
    stuDF.groupBy($"clazz").agg(count($"id") as "cnt").where($"cnt" > 100)
    //      .show()

    // orderBy 排序
    // 统计班级人数 并且按照人数降序排列
    stuDF.groupBy($"clazz").agg(count($"id") as "cnt").orderBy($"cnt".desc)
    //      .show()

    // limit 限制返回的条数
    // 统计班级人数 并且按照人数降序排列 并返回前3名
    stuDF.groupBy($"clazz").agg(count($"id") as "cnt").orderBy($"cnt".desc).limit(3)
    //      .show()

    // union 联接 类似SQL中的union all，不会去重
    val stuSampleDF01: Dataset[Row] = stuDF.sample(withReplacement = false, fraction = 0.01, seed = 1)
    val stuSampleDF02: Dataset[Row] = stuDF.sample(withReplacement = false, fraction = 0.01, seed = 1)
    stuSampleDF01.union(stuSampleDF02)
    //      .show()
    // 如果需要实现SQL中的union操作，即对数据进行去重，则可以使用distinct
    stuSampleDF01.union(stuSampleDF02).distinct()
    //      .show()

    // join 关联
    val scoDF: DataFrame = spark.read.format("csv").option("sep", ",").schema("id String,subject_id String,score Int")
      .load("spark/data/stu/score.txt")

    // 内连接：join / inner join
    stuDF.join(scoDF, "id")
    //      .show() // 默认关联的方式就是内连接
    // 左连接：left join  / left outer join
    stuDF.join(scoDF, List("id"), "left")
    //      .show()
    // 右连接：right join  / right outer join
    stuDF.join(scoDF, List("id"), "right")
    //      .show()
    // 全连接：full join  / full outer join
    stuDF.join(scoDF, List("id"), "full")
      .orderBy($"id")
      .show()


    // 使用列表达式的方式指定关联条件
    //    stuDF.join(scoDF, $"id" === $"id").show() // 当关联的列名相同时，无法直接使用列表达式
    //    stuDF.join(scoDF, $"id" === $"id", "left").show()
    // 可以先进行重命名
    val renameScoDF: DataFrame = scoDF.withColumnRenamed("id", "sid")
    stuDF.join(renameScoDF, $"id" === $"sid")
    //      .show()
    stuDF.join(renameScoDF, $"id" === $"sid", "left")
    //      .show()
    stuDF.join(renameScoDF, $"id" === $"sid", "right")
    //      .show()
    stuDF.join(renameScoDF, $"id" === $"sid", "full")
      .orderBy($"id", $"sid")
      .show()

  }

}
